package com.youku.hot.video;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * @author User 统计一天的数据窗口内，一个视频的vv，捎带分类.
 */
public class StatVVCountInOneDay {
	public static class MapClass extends
			Mapper<LongWritable, VVLogPreHourWritable, Text, LongWritable> {

		@Override
		protected void map(LongWritable key, VVLogPreHourWritable value,
				Context context) throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			long vid = value.getVideo_id();
			String cat = value.getCt();
			int num = value.getNumber();
			if (num != 59) {
				context.getCounter("nostart", "vv").increment(1);
				return;
			}
			String outkey = vid + "\t" + cat;
			context.write(new Text(outkey), new LongWritable(1));
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();
		Job job = new Job(conf, "StatVVCountInOneDay");
		job.setJarByClass(StatVVCountInOneDay.class);
		job.setInputFormatClass(VVLogPreHourWritableInputFormat.class);
		job.setMapperClass(MapClass.class);
		job.setCombinerClass(LongSumReducer.class);
		job.setReducerClass(LongSumReducer.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);
		for (int i = 0; i < otherArgs.length - 1; i++) {
			FileInputFormat.addInputPaths(job, otherArgs[i]);
		}
		FileOutputFormat.setOutputPath(job, new Path(
				otherArgs[otherArgs.length - 1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
